package ba.keco.hbase.versions;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import ba.keco.hbase.lib.HBaseShowcaseInputFormat;
import ba.keco.hbase.lib.Util;

public class MapReduceTest extends Configured implements Tool {

	public static void main(String[] args) throws Exception {
		ToolRunner.run(new MapReduceTest(), args);
	}

	@Override
	public int run(String[] arg0) throws Exception {

		int cfs = getConf().getInt("hbase.showcase.cf.number", -1);
		int versions = getConf().getInt("hbase.showcase.versions", -1);

		boolean keepTable = getConf().getBoolean("hbase.showcase.keep.table", true);

		String tableName = getConf().get("hbase.showcase.table.name");

		try {
			Util.delete(tableName);
		} catch (Exception e) {
		}

		Util.createTableVersions(tableName, cfs, versions);

		getConf().set(TableOutputFormat.OUTPUT_TABLE, tableName);

		Job job = new Job(getConf(), "Versions-test-" + versions);

		job.setJarByClass(MapReduceTest.class);

		job.setNumReduceTasks(0);
		job.setInputFormatClass(HBaseShowcaseInputFormat.class);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.setMapperClass(VersionsMapper.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Put.class);

		long startTime = System.currentTimeMillis();
		boolean isSuccessful = job.waitForCompletion(true);
		long endTime = System.currentTimeMillis();
		System.out.println("###," + versions + "," + (endTime - startTime));
		if (!keepTable) {
			Util.delete(tableName);
		}
		return isSuccessful ? 0 : 1;
	}

	public static class VersionsMapper extends Mapper<NullWritable, NullWritable, NullWritable, Put> {

		private int cfs;
		private int maxversions;

		private int qualifiersPerColumnFamily;

		private HTable table;

		protected void setup(Context context) throws java.io.IOException, InterruptedException {
			cfs = context.getConfiguration().getInt("hbase.showcase.cf.number", -1);
			int qfs = context.getConfiguration().getInt("hbase.showcase.qf.number", -1);
			String tableName = context.getConfiguration().get("hbase.showcase.table.name");
			maxversions = context.getConfiguration().getInt("hbase.showcase.versions", -1);

			table = new HTable(context.getConfiguration(), tableName);

			System.out.println("setup done");
			qualifiersPerColumnFamily = qfs / cfs;
		}

		protected void map(NullWritable key, NullWritable value, Context context) throws java.io.IOException, InterruptedException {
			Put p = Util.generateRecord(cfs, qualifiersPerColumnFamily, maxversions);
			byte[] rowkey =  p.getRow();
			context.write(NullWritable.get(), p);
			Get g = new Get(rowkey);
			g.setMaxVersions(maxversions);
			org.apache.hadoop.hbase.client.Result r = table.get(g);
			System.out.println(r.isEmpty());
		};
	}
}
